import java.io.OutputStream;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class DatabaseConnection {
    private static final String URL = "jdbc:mysql://localhost:3306/ry-vue";
    private static final String USER = "root";
    private static final String PASSWORD = "DzjcMysql@123456.";
    private static final String SERVER_HOST = "192.168.0.200";
    private static final int SERVER_PORT = 9096;
    private static final byte[] FRAME_HEADER = {(byte) 0x57, (byte) 0x53};
    private static final byte FRAME_TYPE_DATA = 0x01;
    private static final long DEVICE_SN = 610100122060001L; // 示例SN码
    private static byte rollChecksum = 0;

    static class KafkaData {
        long epoch;
        float xecef;
        float yecef;
        float zecef;

        KafkaData(long epoch, float xecef, float yecef, float zecef) {
            this.epoch = epoch;
            this.xecef = xecef;
            this.yecef = yecef;
            this.zecef = zecef;
        }
    }

    public static void main(String[] args) {
        Connection connection = null;
        List<KafkaData> dataList = new ArrayList<>();

        try {
            Class.forName("com.mysql.cj.jdbc.Driver");
            connection = DriverManager.getConnection(URL, USER, PASSWORD);
            System.out.println("数据库连接成功！");

            String selectSQL = "SELECT epoch, xecef, yecef, zecef FROM surf_kafkadata WHERE MONITORPOINTCODE = 12";
            PreparedStatement selectStmt = connection.prepareStatement(selectSQL);
            ResultSet resultSet = selectStmt.executeQuery();

            while (resultSet.next()) {
                long epoch = Long.parseLong(resultSet.getString("epoch")); // 假设epoch为字符串
                float xecef = Float.parseFloat(resultSet.getString("xecef"));
                float yecef = Float.parseFloat(resultSet.getString("yecef"));
                float zecef = Float.parseFloat(resultSet.getString("zecef"));
                dataList.add(new KafkaData(epoch, xecef, yecef, zecef));
            }
            System.out.println("读取到 " + dataList.size() + " 条 MONITORPOINTCODE = 12 的数据");

        } catch (ClassNotFoundException e) {
            System.out.println("未找到 JDBC 驱动: " + e.getMessage());
        } catch (SQLException e) {
            System.out.println("数据库操作失败: " + e.getMessage());
        } catch (NumberFormatException e) {
            System.out.println("数据解析失败: " + e.getMessage());
        } finally {
            try {
                if (connection != null && !connection.isClosed()) {
                    connection.close();
                    System.out.println("数据库连接已关闭");
                }
            } catch (SQLException e) {
                System.out.println("关闭连接失败: " + e.getMessage());
            }
        }

        if (!dataList.isEmpty()) {
            ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
            final int[] index = {0};

            scheduler.scheduleAtFixedRate(() -> {
                if (index[0] < dataList.size()) {
                    KafkaData data = dataList.get(index[0]);
                    byte[] packet = packData(data);
                    sendToServer(packet);
                    index[0]++;
                } else {
                    System.out.println("所有数据已发送完毕");
                    scheduler.shutdown();
                }
            }, 0, 3, TimeUnit.SECONDS);
        } else {
            System.out.println("没有数据可发送");
        }
    }

    private static byte[] packData(KafkaData data) {
        int dataItemCount = 3; // X, Y, Z
        int dataItemSize = 6;  // 2字节类型 + 4字节浮点值
        int totalLength = 2 + 1 + 2 + 8 + 8 + 1 + (dataItemCount * dataItemSize) + 1 + 1; // 33字节

        ByteBuffer buffer = ByteBuffer.allocate(totalLength).order(ByteOrder.LITTLE_ENDIAN);

        // 帧头
        buffer.put(FRAME_HEADER);
        // 帧类型
        buffer.put(FRAME_TYPE_DATA);
        // 数据帧长度
        short frameLength = (short) (totalLength - 5); // 28字节
        buffer.putShort(frameLength);
        // 采集设备SN码
        buffer.putLong(DEVICE_SN);
        // 时间戳
        buffer.putLong(data.epoch);
        // 数据项数量
        buffer.put((byte) dataItemCount);

        // 数据项1: X (10101)
        buffer.putShort((short) 10101);
        buffer.putFloat(data.xecef);
        // 数据项2: Y (10102)
        buffer.putShort((short) 10102);
        buffer.putFloat(data.yecef);
        // 数据项3: Z (10103)
        buffer.putShort((short) 10103);
        buffer.putFloat(data.zecef);

        // 滚动校验
        rollChecksum = (byte) (rollChecksum + 1);
        buffer.put(rollChecksum);

        // 累加校验
        byte[] dataSoFar = new byte[buffer.position()];
        buffer.position(0);
        buffer.get(dataSoFar);
        byte checksum = calculateChecksum(dataSoFar);
        buffer.put(checksum);

        return buffer.array();
    }

    private static byte calculateChecksum(byte[] data) {
        int sum = 0;
        for (byte b : data) {
            sum += (b & 0xFF);
        }
        return (byte) (sum & 0xFF);
    }

    private static void sendToServer(byte[] data) {
        try (Socket socket = new Socket(SERVER_HOST, SERVER_PORT);
             OutputStream out = socket.getOutputStream()) {
            out.write(data);
            out.flush();
            System.out.println("已发送到 " + SERVER_HOST + ":" + SERVER_PORT + " 数据包长度: " + data.length + " 字节");
            // 打印字节内容供调试
            for (byte b : data) {
                System.out.printf("%02X ", b);
            }
            System.out.println();
        } catch (Exception e) {
            System.out.println("发送失败: " + e.getMessage());
        }
    }
}